package com.jiuan.transfer.mqtt;
import java.net.ConnectException;
import java.net.NoRouteToHostException;
import java.net.URISyntaxException;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.fusesource.mqtt.client.Future;
import org.fusesource.mqtt.client.FutureConnection;
import org.fusesource.mqtt.client.MQTT;
import org.fusesource.mqtt.client.QoS;

import com.jiuan.transfer.mqtt.domain.Config;
import com.jiuan.transfer.mqtt.domain.UUIDGenerator;

/**
 * @Title: MqttUtil.java
 * @Package com.jiuan.transfer.mqtt
 * @Description: TODO
 * @author Lee
 * @date 2016年8月16日 下午12:22:54
 * @version V1.0
 */

public class MqttUtil {

    private static ConcurrentHashMap<String, FutureConnection> connectionMap = new ConcurrentHashMap();
    private static ConcurrentHashMap<String, MQTT> mqttMap = new ConcurrentHashMap();
    private static Set<String> errorIp = new HashSet<String>();


    private static synchronized boolean errorIp(String ip){
        FutureConnection c=null ;
        try {
            c = connectionMap.remove(ip);
            c.disconnect();
            c.kill();
            c=null;
            System.out.println("destroy connection: "+ip);
            errorIp.add(ip);
        } catch (Exception e) {

            e.printStackTrace();
        }
        return null==c;
    }

    @SuppressWarnings("unused")
    private static synchronized boolean isErrorIp(String ip) {
        return errorIp.contains(ip);
    }

    public static synchronized void clearAll() {
        for (FutureConnection item : connectionMap.values()) {
            item.kill();
        }
        connectionMap.clear();
        mqttMap.clear();
        errorIp.clear();
    }

    public static synchronized MQTT createMqtt(String ip) throws Exception {
        //阻塞等待线程 进入时再次检查，避免多次创建
        if(mqttMap.containsKey(ip))
            return mqttMap.get(ip);
        System.out.println("createMqtt: "+ip);
        MQTT mqtt = new MQTT();
        try {
            mqtt.setHost(ip, Config.getMqtt_port());
        } catch (URISyntaxException e) {
            System.err.println("mqtt 创建失败,检查ip,port！ ip:"+ip+"    port:"+Config.getMqtt_port());
            throw new Exception("mqtt client 创建失败");
        }
        mqtt.setClientId(Config.getMqtt_clientid_prefix()+UUIDGenerator.generateUUIDString(50)); // 用于设置客户端会话的ID。在setCleanSession(false);被调用时，MQTT服务器利用该ID获得相应的会话。此ID应少于23个字符，默认根据本机地址、端口和时间自动生成
        mqtt.setCleanSession(true); // 若设为false，MQTT服务器将持久化客户端会话的主体订阅和ACK位置，默认为true
        mqtt.setKeepAlive(Config.getMqtt_keepalive());// 定义客户端传来消息的最大时间间隔秒数，服务器可以据此判断与客户端的连接是否已经断开，从而避免TCP/IP超时的长时间等待
        mqtt.setVersion("3.1.0");
        // 失败重连接设置说明
        mqtt.setConnectAttemptsMax(10L);// 客户端首次连接到服务器时，连接的最大重试次数，超出该次数客户端将返回错误。-1意为无重试上限，默认为-1
        mqtt.setReconnectAttemptsMax(3L);// 客户端已经连接到服务器，但因某种原因连接断开时的最大重试次数，超出该次数客户端将返回错误。-1意为无重试上限，默认为-1
        mqtt.setReconnectDelay(10L);// 首次重连接间隔毫秒数，默认为10ms
        mqtt.setReconnectDelayMax(30000L);// 重连接间隔毫秒数，默认为30000ms
        mqtt.setReconnectBackOffMultiplier(2);// 设置重连接指数回归。设置为1则停用指数回归，默认为2
        // Socket设置说明
        mqtt.setReceiveBufferSize(65536);// 设置socket接收缓冲区大小，默认为65536（64k）
        mqtt.setSendBufferSize(65536);// 设置socket发送缓冲区大小，默认为65536（64k）
        mqtt.setTrafficClass(8);// 设置发送数据包头的流量类型或服务类型字段，默认为8，意为吞吐量最大化传输
        // 带宽限制设置说明
        mqtt.setMaxReadRate(0);// 设置连接的最大接收速率，单位为bytes/s。默认为0，即无限制
        mqtt.setMaxWriteRate(0);// 设置连接的最大发送速率，单位为bytes/s。默认为0，即无限制
        mqttMap.put(ip, mqtt);
        return mqtt;
    }


    private static synchronized FutureConnection  createConnection(String ip) throws Exception {
        if(connectionMap.containsKey(ip)){
            return connectionMap.get(ip);
        }
        System.out.println("createConnection: "+ip);
        FutureConnection connection=null;
        if(mqttMap.containsKey(ip)){
            connection = mqttMap.get(ip).futureConnection();
        }else{
            connection = createMqtt(ip).futureConnection();
        }

        Future<Void> future = connection.connect();
        future.await(2, TimeUnit.SECONDS);

        connectionMap.put(ip, connection);
        errorIp.remove(ip);
        return connection;
    }

    public static void publish(String ip,String topic, byte[] message) throws Exception{
        try {
            FutureConnection connection = getConnection(ip);
            Future<Void> future = connection.publish(topic, message,
                    QoS.AT_LEAST_ONCE, false);
            future.await(1, TimeUnit.SECONDS);
        } catch (ConnectException | TimeoutException | NoRouteToHostException e){
            synchronized (MqttUtil.class) {
                if(!isErrorIp(ip)){
                    System.err.println("MQTT 连接中断！\t"+e.getMessage());
                    MqttUtil.errorIp(ip);
                    //重发
                    publish(ip,topic,message);
                };

            }
        }

    }

    private static FutureConnection  getConnection(String ip) throws Exception {

        if(connectionMap.containsKey(ip)){
            return connectionMap.get(ip);
        }else{
            return createConnection(ip);
        }
    }

    @SuppressWarnings("unused")
    private static void connect(FutureConnection connection) throws Exception {
        if (!connection.isConnected()) {
            System.out.print("try connect... ");
            Future<Void> future = connection.connect();
            future.await(1, TimeUnit.SECONDS);
            System.out.println("result: "+connection.isConnected());
        }
    }

    @SuppressWarnings("unused")
    private static void close(FutureConnection connection) {
        if (connection.isConnected()) {
            connection.disconnect();
        }
    }


    public static boolean isConnectOK(FutureConnection connection) {
        try {
            System.out.print("Check is Connect OK:\t");
            Future<Void> future = connection.publish("Topic_test_connect", "test_connect".getBytes(),
                    QoS.AT_LEAST_ONCE, false);
            future.await(2, TimeUnit.SECONDS);
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
        return true;
    }





    public static void main(String[] args) throws Exception {

        Config.initConfig("");

        FutureConnection connection = createConnection("192.168.12.193");
        System.out.println("isconn: "+connection.isConnected());
        System.out.println(isConnectOK(connection));;

        System.out.println("kill");
        connection.kill();
        System.out.println("isconn: "+connection.isConnected());
        System.out.println(isConnectOK(connection));;
        Thread.sleep(1000);


        System.out.println("dis\n");
        Future<Void> future = connection.disconnect();
        future.await(1, TimeUnit.SECONDS);
        System.out.println("isconn: "+connection.isConnected());
        Thread.sleep(1000);


        System.out.println("conn");
        connect(connection);
        System.out.println("isconn: "+connection.isConnected());
        System.out.println(isConnectOK(connection));;
        Thread.sleep(1000);


        System.out.println("conn");
        connect(connection);
        System.out.println("isconn: "+connection.isConnected());
        System.out.println(isConnectOK(connection));;
        Thread.sleep(1000);

        System.out.println("kill");
        connection.kill();
        System.out.println("isconn: "+connection.isConnected());
        System.out.println(isConnectOK(connection));;
        Thread.sleep(1000);


        System.out.println("res\n");
        connection.resume();
        System.out.println("isconn: "+connection.isConnected());
        System.out.println(isConnectOK(connection));

		/*


		try {
			for (int i = 0; i < 100; i++) {
				try {
					publish("192.168.12.193", "TOPIC_001", "test0000001".getBytes());
				} catch (Exception e) {
					System.out.println("ex:for pub");
					e.printStackTrace();
					connectionMap.remove("192.168.12.193");
				}
				Thread.sleep(3000);
			}

		} catch (Exception e) {
			System.out.println("=");
			e.printStackTrace();
		}

		*/
    }
}
